package com.qf.efserver.track.consumer;

import com.alibaba.fastjson.JSON;
import com.qf.common.redis.utils.RedisUtils;
import com.qf.efserver.track.param.TrackLogParam;
import com.qf.efserver.track.service.MotionTrackService;
import com.qf.efserver.track.vo.TrackVO;
import com.rabbitmq.client.Channel;
import org.springframework.amqp.core.Message;
import org.springframework.amqp.rabbit.annotation.RabbitHandler;
import org.springframework.amqp.rabbit.annotation.RabbitListener;
import org.springframework.stereotype.Component;

import javax.annotation.Resource;
import java.io.IOException;
import java.util.List;
import java.util.concurrent.ThreadPoolExecutor;
import java.util.stream.Collectors;

/**
 * Author hehecheng
 * Date 2024/11/18
 * 祝我们都能遇见真诚的人
 * 见字如面
 */
@Component
@RabbitListener(queues = "track_log")
public class TrackLogConsumer {

    @Resource
    private MotionTrackService motionTrackService;

    @Resource
    private ThreadPoolExecutor threadPoolExecutor;

    /**
     * 对完成运动的运动轨迹存入到数据库中
     *
     * @param msg
     * @param channel
     * @param message
     */
    @RabbitHandler
    public void receiveACK(String msg, Channel channel, Message message) {

        try {
            TrackLogParam trackLogParam = JSON.parseObject(msg, TrackLogParam.class);
            String key2 = String.format("consumer:%s", trackLogParam.getActivityId());
            boolean ret = RedisUtils.hasKey(key2);
            if (!ret) {
                String key = String.format("activity:%s:uid:%s", trackLogParam.getActivityId(), trackLogParam.getUid());
                List<Object> list = RedisUtils.listGet(key, 0, -1);
                List<TrackVO> trackVOS = list.stream().map(item -> (TrackVO) item).collect(Collectors.toList());
                motionTrackService.batchInsertion(trackVOS);
                RedisUtils.setValueTimeout(key2, 1, 10);
            }
            // 第一个参数：是内部生成的消息唯一标记，
            // 第二个参数：表示一个都送确认一个消息，不能批量确认
            channel.basicAck(message.getMessageProperties().getDeliveryTag(), false);
        } catch (IOException e) {
            e.printStackTrace();
            try {
                // 如果业务出现问题，拒绝消息的处理
                // 第二个参数表示，是否出现进入队列
                // TODO 补偿机制
                channel.basicReject(message.getMessageProperties().getDeliveryTag(), false);
            } catch (IOException ex) {
                ex.printStackTrace();
            }
        }
    }
}
